package com.personas.analysic.analyserv.task;

import com.personas.analysic.analyserv.entity.YearBase;
import com.personas.analysic.analyserv.map.YearBaseMap;
import com.personas.analysic.analyserv.reduce.YearBaseReduce;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;

import java.util.List;

public class YearBaseTask {

    public static void main(String[] args) {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> stringDataSource = env.readTextFile("");
        DataSet<YearBase> mapSet  = stringDataSource.map(new YearBaseMap());
        DataSet<YearBase> reduceSet = mapSet.groupBy("group").reduce(new YearBaseReduce());
        try {
            List<YearBase> collect = reduceSet.collect();
            for (YearBase yearBase : collect) {
                String year = yearBase.getYear();
                int count = yearBase.getCount();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }



}
